
import os
import time
import unittest
import uuid

import redis

'''
4.1.1 处理日志快照
记录日志函数process_logs：将处理进度记录到redis里面，程序在崩溃之后，根据记录继续执行之前未完成的工作。
1）参数：conn(redis连接)，path(日志路径)，callback(日志文件函数)
2）获取当前文件的进度current_file，偏移量offset；
3）跟新进度（嵌套函数/pipline);
4）遍历所有的文件os.listdir(path)
5）略过已经处理的文件
6）处理系统崩溃未能处理的日志文件（inp.seek）
7）遍历文件行数据并且枚举(enumerate)
8）更新当前文件的进度和偏移量
'''
def process_logs(conn, path, callback):
    current_file, offset = conn.mget('progress:file', 'progress:position')

    pipe = conn.pipeline()

    # 定义内部函数
    def update_progress():
        pipe.mset({
            'progress:file': fname,
            'progress:position': offset
        })
        pipe.execute()

    for fname in sorted(os.listdir(path)):
        if fname < current_file:
            continue
        # 打开文件
        inp = open(os.path.join(path, fname), 'rb')
        if fname == current_file:
            inp.seek(int(offset, 10))
        else:
            offset = 0

        current_file = None

        for lno, line in enumerate(inp):
            callback(pipe, line)
            offset = int(offset) + len(line)
            # 每处理1000行更新一次日志记录
            if not (lno+1) % 1000:
                update_progress()
        update_progress()
        # 关闭文件
        inp.close()

'''
4.2.4 检查硬盘写入
1）将令牌添加到主服务器mconn.zadd
2）等待从服务器完成同步
3）等待从服务器接收数据更新
4）最多等待1秒
5）检查数据同步是否更新到硬盘
6）清除令牌
'''
def wait_for_sync(mconn, sconn):
    identifier = str(uuid.uuid4())
    mconn.zadd('sync:wait', {identifier: time.time()})
    # 等待从服务器完成同步
    while not sconn.info()['master_link_status'] != 'up':
        time.sleep(.001)

    while not sconn.zscore('sync:wait', identifier):
        time.sleep(.001)
    # 最多等待1秒
    deadline = time.time() + 1.01
    while time.time() < deadline:
        # 检查数据同步是否更新到硬盘
        if sconn.info()['aof_pending_bio_fsync'] == 0:
            break
        time.sleep(.001)

    mconn.zrem('sync:wait', identifier)
    mconn.zremrangebyscore('sync:wait', 0, time.time()-900)

'''
4.4.2 将商品放到市场，用事务监控
1）监视库存的变化pipe.watch(inventory)；
2）检查用户是否持有被销售的产品；
3）如果商品非库存，停止监视，返回None;
4）如果商品在库存，添加到市场market里面；
5）如果pipe.execute()无异常，则事务执行成功，监视结束；
5）如果pipe.execute()有异常，则事务执行失败，重试(pass，继续执行while)；
'''
def list_item(conn, itemid, sellerid, price):
    inventory = "inventory:%s"%sellerid
    item = "%s.%s"%(itemid, sellerid)
    end = time.time() + 5
    pipe = conn.pipeline()

    # 出现错误时，重试的时间为5秒
    while time.time() < end:
        try:
            pipe.watch(inventory)
            if not pipe.sismember(inventory, itemid):
                pipe.unwatch()
                return None

            pipe.multi()
            pipe.zadd("market:", {item: price})
            pipe.srem(inventory, itemid)
            pipe.execute()
            return True
        except redis.exceptions.WatchError:
            pass
    return False

'''
4.4.3 从市场上购买商品，用事务监控
1）监控商品信息、用户信息pipe.watch；
2）检查商品信息是否变化，检查买家是否有足够的钱；
3）将钱支付给卖家，将商品转移给买家；
4）如果商品信息、用户信息在交易过程中出现变化 ，则重试（pass）
'''
def purchase_item(conn, buyerid, itemid, sellerid, lprice):
    buyer = "users:%s"%buyerid
    seller = "users:%s"%sellerid
    item = "%s.%s"%(itemid, sellerid)
    inventory = "inventory:%s"%buyerid
    end = time.time() + 10
    pipe = conn.pipeline()

    # 出现错误时，重试的时间为10秒
    while time.time() < end:
        try:
            pipe.watch("market:", buyer)

            price = pipe.zscore("market:", item)
            funds = int(pipe.hget(buyer, "funds"))
            if price != lprice or price > funds:
                pipe.unwatch()
                return None

            pipe.multi()
            pipe.hincrby(seller, "funds", int(price))
            pipe.hincrby(buyer, "funds", int(-price))
            pipe.sadd(inventory, itemid)
            pipe.zrem("market:", item)
            pipe.execute()
            return True
        except redis.exceptions.WatchError:
            pass
    return False

'''
4.5 非事务流水线pipeline(False)，减少通信次数，提高性能
'''
def update_token_pipeline(conn, token, user, item=None):
    timestamp = time.time()
    # 传入False为不使用事务，True或为空时启用事务
    pipe = conn.pipeline(False)
    pipe.hset('login:', token, user)
    pipe.zadd('recent:', {token: timestamp})
    if item:
        pipe.zadd('viewed:' + token, {item: timestamp})
        pipe.zremrangebyrank('viewed:' + token, 0, -26)
        pipe.zincrby('viewed:', -1, item)
    pipe.execute()


# 更新token
def update_token(conn, token, user, item=None):
    timestamp = time.time()
    conn.hset('login:', token, user)
    conn.zadd('recent:', token, timestamp)
    if item:
        key = 'viewed:' + token
        conn.lrem(key, item)
        conn.rpush(key, item)
        conn.ltrim(key, -25, -1)
        conn.zincrby('viewed:', item, -1)

# 验证性能
def benchmark_update_token(conn, duration):
    # 函数名放在元组里面
    for function in (update_token, update_token_pipeline):
        count = 0
        start = time.time()
        end = start + duration
        while time.time() < end:
            count += 1
            # 循环调用函数
            function(conn, 'token', 'user', 'item')
        delta = time.time() - start
        # 打印执行的时间
        print(function.__name__, count, delta, count / delta)

#--------------- Below this line are helpers to test the code ----------------

class TestCh04(unittest.TestCase):
    def setUp(self):
        import redis
        self.conn = redis.Redis(host="localhost", db=15)
        self.conn.flushdb()

    def tearDown(self):
        self.conn.flushdb()
        del self.conn
        print()
        print()

    # We can't test process_logs, as that would require writing to disk, which
    # we don't want to do.

    # We also can't test wait_for_sync, as we can't guarantee that there are
    # multiple Redis servers running with the proper configuration

    def test_list_item(self):
        import pprint
        conn = self.conn

        print("We need to set up just enough state so that a user can list an item")
        seller = 'userX'
        item = 'itemX'
        conn.sadd('inventory:' + seller, item)
        i = conn.smembers('inventory:' + seller)
        print("The user's inventory has:", i)
        self.assertTrue(i)
        print()

        print("Listing the item...")
        l = list_item(conn, item, seller, 10)
        print("Listing the item succeeded?", l)
        self.assertTrue(l)
        r = conn.zrange('market:', 0, -1, withscores=True)
        print("The market contains:")
        pprint.pprint(r)
        self.assertTrue(r)
        self.assertTrue(any(x[0] == b'itemX.userX' for x in r))

    def test_purchase_item(self):
        self.test_list_item()
        conn = self.conn
        
        print("We need to set up just enough state so a user can buy an item")
        buyer = 'userY'
        conn.hset('users:userY', 'funds', 125)
        r = conn.hgetall('users:userY')
        print("The user has some money:", r)
        self.assertTrue(r)
        self.assertTrue(r.get(b'funds'))
        print()

        print("Let's purchase an item")
        p = purchase_item(conn, 'userY', 'itemX', 'userX', 10)
        print("Purchasing an item succeeded?", p)
        self.assertTrue(p)
        r = conn.hgetall('users:userY')
        print("Their money is now:", r)
        self.assertTrue(r)
        i = conn.smembers('inventory:' + buyer)
        print("Their inventory is now:", i)
        self.assertTrue(i)
        self.assertTrue(b'itemX' in i)
        self.assertEqual(conn.zscore('market:', 'itemX.userX'), None)

    def test_benchmark_update_token(self):
        benchmark_update_token(self.conn, 5)

if __name__ == '__main__':
    unittest.main()
